package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.bean.ExamPaperStatistics;
import com.atguigu.edu.realtime.util.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsExaminationPaperExamWindow {
    public static void main(String[] args) {
        //1 创建环境
        StreamExecutionEnvironment env = EnvAndCheckUtils.getEnvAndMarkCheck();
        //2 获取Kafka数据源
        String topic = "dwd_test_exam";
        String groupId = "DwsExaminationPaperExamWindow";
        KafkaSource<String> kafkaSource = KafkaUtils.getKafkaSource(topic, groupId);
        DataStreamSource<String> sourceDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //sourceDs.print(">>>>>>");
        //3 数据格式转换str -> ExamPaperStatistics
        SingleOutputStreamOperator<ExamPaperStatistics> processDS = sourceDs.process(new ProcessFunction<String, ExamPaperStatistics>() {
            @Override
            public void processElement(String str, ProcessFunction<String, ExamPaperStatistics>.Context context, Collector<ExamPaperStatistics> collector) throws Exception {
                JSONObject jsonObj = JSON.parseObject(str);
                Long ts = jsonObj.getLong("ts") * 1000L;
                Double score = jsonObj.getDouble("score");
                collector.collect(ExamPaperStatistics.builder()
                                .paperId(jsonObj.getString("paper_id"))
                                .examTotalScore((long)score.doubleValue())
                                .examTakenCount(1L)
                                .examTotalDuringSec(jsonObj.getLong("duration_sec"))
                                .ts(ts)
                        .build());
            }
        });
        processDS.print("格式转换：");
        //4 设置水位线
        SingleOutputStreamOperator<ExamPaperStatistics> watermarksDS = processDS
                .assignTimestampsAndWatermarks(
                WatermarkStrategy.<ExamPaperStatistics>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ExamPaperStatistics>() {
                            @Override
                            public long extractTimestamp(ExamPaperStatistics eps, long l) {
                                return eps.getTs();
                            }
                        })
        );
        //watermarksDS.print(">>>>>>>>");
        // 分组
        KeyedStream<ExamPaperStatistics, String> keyByDs = watermarksDS.keyBy(new KeySelector<ExamPaperStatistics, String>() {
            @Override
            public String getKey(ExamPaperStatistics examPaperStatistics) throws Exception {
                return examPaperStatistics.getPaperId();
            }
        });

        //5 开窗
        WindowedStream<ExamPaperStatistics, String, TimeWindow> windowDS = keyByDs.window(TumblingEventTimeWindows.of(Time.seconds(10L)));
        //AllWindowedStream<ExamPaperStatistics, TimeWindow> windowAllDS = watermarksDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(3)));

        //6 聚合
        SingleOutputStreamOperator<ExamPaperStatistics> reduceDs = windowDS.reduce(
                new ReduceFunction<ExamPaperStatistics>() {
                    @Override
                    public ExamPaperStatistics reduce(ExamPaperStatistics t1, ExamPaperStatistics t2) throws Exception {
                        t1.setExamTotalDuringSec(t1.getExamTotalDuringSec() + t2.getExamTotalDuringSec());
                        t1.setExamTotalScore(t1.getExamTotalScore() + t2.getExamTotalScore());
                        t1.setExamTakenCount(t1.getExamTakenCount() + t2.getExamTakenCount());
                        return t1;
                    }
                },
                new ProcessWindowFunction<ExamPaperStatistics, ExamPaperStatistics, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<ExamPaperStatistics, ExamPaperStatistics, String, TimeWindow>.Context context, Iterable<ExamPaperStatistics> iterable, Collector<ExamPaperStatistics> collector) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String date = DateFormatUtil.toDate(context.window().getStart());
                        for (ExamPaperStatistics element : iterable) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setCur_date(date);
                            //element.setTs(System.currentTimeMillis());
                            collector.collect(element);
                        }
                    }
                }
        );
        reduceDs.print("聚合：");
        SingleOutputStreamOperator<ExamPaperStatistics> mapDS = reduceDs.map(
                new RichMapFunction<ExamPaperStatistics, ExamPaperStatistics>() {
                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = HbaseUtil.getHbaseConnect();
                    }

                    @Override
                    public void close() throws Exception {
                        HbaseUtil.closeHbaseConnect(conn);
                    }

                    @Override
                    public ExamPaperStatistics map(ExamPaperStatistics Bean) throws Exception {
                        //获取要关联的维度的主键
                        String skuId = Bean.getPaperId();
                        //根据主键获取维度对象
                        JSONObject dimJsonObj = DimUtil.getDimInfoOutRedis(conn, EduConfig.HBASE_NAMESPACE, "dim_test_paper", skuId);
                        System.out.println(dimJsonObj);
                        //将维度对象的属性补充到流中对象上
                        Bean.setPaperTitle(dimJsonObj.getString("paper_title"));
                        Bean.setCourseId(dimJsonObj.getString("course_id"));
                        return Bean;
                    }
                }
        );
        mapDS.print("napDS:");
        //7 关联试卷维度表
        SingleOutputStreamOperator<ExamPaperStatistics> paperReduceDS = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<ExamPaperStatistics>("dim_test_paper") {
                    @Override
                    public void join(ExamPaperStatistics statistics, JSONObject dimJsonObj) {
                        statistics.setPaperTitle(dimJsonObj.getString("paper_title"));
                        statistics.setCourseId(dimJsonObj.getString("course_id"));
                    }

                    @Override
                    public String getKey(ExamPaperStatistics statistics) {
                        return statistics.getPaperId();
                    }
                },
                60, TimeUnit.SECONDS
        );
        //paperReduceDS.print("paper:");
        //8 关联课程维度表
        SingleOutputStreamOperator<ExamPaperStatistics> paperCourseReduceDS = AsyncDataStream.unorderedWait(
                paperReduceDS,
                new DimAsyncFunction<ExamPaperStatistics>("dim_course_info") {
                    @Override
                    public void join(ExamPaperStatistics statistics, JSONObject dimJsonObj) {
                        statistics.setCourseName(dimJsonObj.getString("course_name"));
                    }

                    @Override
                    public String getKey(ExamPaperStatistics statistics) {
                        return statistics.getCourseId();
                    }
                },
                60, TimeUnit.SECONDS
        );
//        paperCourseReduceDS.print("关联结果：");
//        ExamPaperStatistics(stt=2023-09-11 15:47:56, edt=2023-09-11 15:47:58, paperId=1115, paperTitle=JavaWeb从入门到精通测试,
//         courseId=193, courseName=JavaWeb从入门到精通, examTakenCount=1, examTotalScore=25, examTotalDuringSec=821, ts=1694418496404)

        paperCourseReduceDS.map(
                new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_examination_paper_exam_window"));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
